{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "EnergyDisaggregationDemo_Client.ipynb",
      "version": "0.3.2",
      "provenance": [],
      "collapsed_sections": []
    },
    "kernelspec": {
      "display_name": "Python 2",
      "language": "python",
      "name": "python2"
    }
  },
  "cells": [
    {
      "metadata": {
        "id": "m9wE6Zw4-oub",
        "colab_type": "text"
      },
      "cell_type": "markdown",
      "source": [
        "# License\n",
        "\n",
        "Copyright 2019 Google LLC\n",
        "\n",
        "Licensed under the Apache License, Version 2.0 (the \"License\");  \n",
        "you may not use this file except in compliance with the License.  \n",
        "You may obtain a copy of the License at . \n",
        "\n",
        "      http://www.apache.org/licenses/LICENSE-2.0\n",
        "\n",
        "Unless required by applicable law or agreed to in writing,  \n",
        "software distributed under the License is distributed on an \"AS IS\" BASIS,  \n",
        "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  \n",
        "See the License for the specific language governing permissions and  \n",
        "limitations under the License."
      ]
    },
    {
      "metadata": {
        "colab_type": "text",
        "id": "T6dXb-4Luvam"
      },
      "cell_type": "markdown",
      "source": [
        "# Pre-work"
      ]
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "VbkK2gFfm9Ht",
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Upload files (skip this if this is run locally)\n",
        "\n",
        "# Use this cell to update the following files\n",
        "#   1. requirements.txt\n",
        "#   2. e2e_demo_credential.json\n",
        "#   3. roots.pem\n",
        "#   4. rs256.key\n",
        "from google.colab import files\n",
        "uploaded = files.upload()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "s-5P4SKvm7Dh",
        "scrolled": false,
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Install missing packages\n",
        "\n",
        "# run this cell to install packages if some are missing\n",
        "!pip install -r requirements.txt"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "zB_AzB1Mm7Dm",
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Import libraries\n",
        "\n",
        "import json\n",
        "import os\n",
        "import gcsfs\n",
        "import numpy as np\n",
        "import pandas as pd\n",
        "import time\n",
        "import google.cloud.bigquery\n",
        "import datetime\n",
        "import random\n",
        "import ssl\n",
        "import jwt\n",
        "import paho.mqtt.client as mqtt\n",
        "import pandas as pd\n",
        "import json\n",
        "import sys"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "PgvqcvTbm7Dq",
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Configurations\n",
        "\n",
        "# project related\n",
        "GOOGLE_CLOUD_PROJECT = 'your-google-project-id' #@param\n",
        "GOOGLE_APPLICATION_CREDENTIALS = 'e2e_demo_credential.json' #@param\n",
        "os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = GOOGLE_APPLICATION_CREDENTIALS\n",
        "\n",
        "# IoT related\n",
        "REGION = 'us-central1' #@param\n",
        "REGISTRY_ID = 'your-IoT-registry-id' #@param\n",
        "DEVICE_IDS = ['your-device-id1', 'your-device-id2', ...] #@param\n",
        "PRIVATE_KEY_PATH = 'rs256.key' #@param\n",
        "CA_CERTIFICATES_PATH = 'roots.pem' #@param\n",
        "\n",
        "# data related\n",
        "POWER_DATA_PATH = 'gs://gcp_blog/e2e_demo/test.csv'\n",
        "DATASET_ID = 'EnergyDisaggregation'"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "colab_type": "text",
        "id": "uBWiiYeKm7D2"
      },
      "cell_type": "markdown",
      "source": [
        "# Start to publish data"
      ]
    },
    {
      "metadata": {
        "colab_type": "text",
        "id": "pY7Xf0BPm7D9"
      },
      "cell_type": "markdown",
      "source": [
        "When the following cells are run, the program mimics multiple electric meters each of which sends readings at the specified interval to a server residing in GAE.  \n",
        "The GAE forwards the readings to CMLE where a trained model is deployed, and it publishes the data together with CMLE's predictions to pub/sub, both the data and the predictions are also saved in BigQuery for future analysis."
      ]
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "Nj9SynKJm7D2",
        "scrolled": true,
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Utility\n",
        "\n",
        "def load_data(project, path):\n",
        "  \"\"\"Load test data from GCS.\n",
        "\n",
        "  Load test csv data from GCS.\n",
        "\n",
        "  Returns:\n",
        "    pandas.DataFrame, active power data.\n",
        "  \"\"\"\n",
        "  fs = gcsfs.GCSFileSystem(project)\n",
        "  with fs.open(path) as f:\n",
        "    power_data = pd.read_csv(f, index_col=0)['gross']\n",
        "  return power_data\n",
        "\n",
        "# The maximum backoff time before giving up, in seconds.\n",
        "MAXIMUM_BACKOFF_TIME = 32\n",
        "\n",
        "class Publisher:\n",
        "    \"\"\"Publish power data every n seconds.\"\"\"\n",
        "\n",
        "    # The initial backoff time after a disconnection occurs, in seconds.\n",
        "    minimum_backoff_time = 1\n",
        "\n",
        "    # Whether to wait with exponential backoff before publishing.\n",
        "    should_backoff = False\n",
        "\n",
        "    def __init__(self,\n",
        "            project_id,\n",
        "            region,\n",
        "            registry_id,\n",
        "            device_id,\n",
        "            private_key_path,\n",
        "            jwt_algorithm,\n",
        "            jwt_exp_mins,\n",
        "            ca_certs_path,\n",
        "            mqtt_bridge_hostname,\n",
        "            mqtt_bridge_port,\n",
        "            seq_len,\n",
        "            sub_topic='events',\n",
        "            interval=1):\n",
        "        self.project_id = project_id\n",
        "        self.region = region\n",
        "        self.registry_id = registry_id\n",
        "        self.device_id = device_id\n",
        "        self.private_key_path = private_key_path\n",
        "        self.jwt_algorithm = jwt_algorithm\n",
        "        self.jwt_exp_mins = jwt_exp_mins\n",
        "        self.ca_certs_path = ca_certs_path\n",
        "        self.mqtt_bridge_hostname = mqtt_bridge_hostname\n",
        "        self.mqtt_bridge_port = mqtt_bridge_port\n",
        "        self.seq_len=seq_len\n",
        "        self.sub_topic = sub_topic\n",
        "        self.interval = interval\n",
        "\n",
        "        self._count = 0\n",
        "        self._data = {}\n",
        "        self._mqtt_topic = '/devices/{}/{}'.format(self.device_id, self.sub_topic)\n",
        "        self._jwt_iat = datetime.datetime.utcnow()\n",
        "        self._active = False\n",
        "\n",
        "\n",
        "    def create_jwt(self):\n",
        "        \"\"\"Creates a JWT (https://jwt.io) to establish an MQTT connection.\n",
        "            Returns:\n",
        "                An MQTT generated from the given project_id and private key, which\n",
        "                expires in 20 minutes. After 20 minutes, your client will be\n",
        "                disconnected, and a new JWT will have to be generated.\n",
        "            Raises:\n",
        "                ValueError: If the private_key_file does not contain a known key.\n",
        "            \"\"\"\n",
        "        token = {\n",
        "                # The time that the token was issued at\n",
        "                'iat': datetime.datetime.utcnow(),\n",
        "                # The time the token expires.\n",
        "                'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=60),\n",
        "                # The audience field should always be set to the GCP project id.\n",
        "                'aud': self.project_id\n",
        "        }\n",
        "\n",
        "        # Read the private key file.\n",
        "        with open(self.private_key_path, 'r') as f:\n",
        "            private_key = f.read()\n",
        "\n",
        "        print('Creating JWT using {} from private key file {}'.format(\n",
        "                self.jwt_algorithm, self.private_key_path))\n",
        "\n",
        "        return jwt.encode(token, private_key, algorithm=self.jwt_algorithm)\n",
        "\n",
        "    def get_client(self):\n",
        "        \"\"\"Create our MQTT client. The client_id is a unique string that identifies\n",
        "        this device. For Google Cloud IoT Core, it must be in the format below.\"\"\"\n",
        "        client = mqtt.Client(\n",
        "                client_id=('projects/{}/locations/{}/registries/{}/devices/{}'\n",
        "                           .format(\n",
        "                                   self.project_id,\n",
        "                                   self.region,\n",
        "                                   self.registry_id,\n",
        "                                   self.device_id)))\n",
        "\n",
        "        # With Google Cloud IoT Core, the username field is ignored, and the\n",
        "        # password field is used to transmit a JWT to authorize the device.\n",
        "        client.username_pw_set(\n",
        "                username='unused',\n",
        "                password=self.create_jwt())\n",
        "\n",
        "        # Enable SSL/TLS support.\n",
        "        client.tls_set(ca_certs=self.ca_certs_path, tls_version=ssl.PROTOCOL_TLSv1_2)\n",
        "\n",
        "        # Register message callbacks. https://eclipse.org/paho/clients/python/docs/\n",
        "        # describes additional callbacks that Paho supports.\n",
        "        client.on_connect = self.on_connect\n",
        "        client.on_publish = self.on_publish\n",
        "        client.on_disconnect = self.on_disconnect\n",
        "        \n",
        "        return client\n",
        "\n",
        "    def publish(self):\n",
        "        # Finish if inactive\n",
        "        if not self._active:\n",
        "            return\n",
        "\n",
        "        # Process network events.\n",
        "        self.client.loop()\n",
        "\n",
        "        # Wait if backoff is required.\n",
        "        if Publisher.should_backoff:\n",
        "            # If backoff time is too large, give up.\n",
        "            if Publisher.minimum_backoff_time > MAXIMUM_BACKOFF_TIME:\n",
        "                print('Exceeded maximum backoff time. Giving up.')\n",
        "                return\n",
        "            # Otherwise, wait and connect again.\n",
        "            delay = Publisher.minimum_backoff_time + random.randint(0, 1000) / 1000.0\n",
        "            print('Waiting for {} before reconnecting.'.format(delay))\n",
        "            time.sleep(delay)\n",
        "            Publisher.minimum_backoff_time *= 2\n",
        "            self.client.connect(self.mqtt_bridge_hostname, self.mqtt_bridge_port)\n",
        "\n",
        "        # Refresh token if JWT iat has been expired.\n",
        "        seconds_since_issue = (datetime.datetime.utcnow() - self._jwt_iat).seconds\n",
        "        if seconds_since_issue > 60 * self.jwt_exp_mins:\n",
        "            print('Refreshing token after {}s').format(seconds_since_issue)\n",
        "            self._jwt_iat = datetime.datetime.utcnow()\n",
        "            self.client = self.get_client()\n",
        "\n",
        "        # Generate payload\n",
        "        d, t = self._data[self._count]\n",
        "        Publisher.rotate_message(self._msg, d, t)\n",
        "        payload = json.dumps(self._msg).encode('utf-8')\n",
        "\n",
        "        # Publish \"payload\" to the MQTT topic. qos=1 means at least once\n",
        "        # delivery. Cloud IoT Core also supports qos=0 for at most once\n",
        "        # delivery.\n",
        "        self.client.publish(self._mqtt_topic, payload, qos=1)\n",
        "        print('Published: #{0:03d} on {1} sent: ({2}, {3}).'.format(self._count+1, self.device_id, self._msg['power'][-1], self._msg['timestamp'][-1]))\n",
        "        self._count += 1\n",
        "\n",
        "    def start(self, data, start_time=None, cnt=0):\n",
        "        \"\"\"Publish payloads to the MQTT topic.\n",
        "            Args:\n",
        "             data: pandas.DataFrame, active power data.\n",
        "             interval: int, time in second.\n",
        "             start_time: str, publish data collected from start_time.\n",
        "             cnt: int, number of data records to publish. If cnt<=0, publish all.\n",
        "            \"\"\"\n",
        "        data = Publisher.trim_data(data, start_time)\n",
        "        self.client = self.get_client()\n",
        "\n",
        "        data_info = data.values.tolist()\n",
        "        time_info = data.index.values.tolist()\n",
        "        self._data = zip(data_info[self.seq_len-1:], time_info[self.seq_len-1:])\n",
        "        self._msg = {'device_id': self.device_id, 'power': [0] + data_info[:self.seq_len-1],\n",
        "            'timestamp': [0] + time_info[:self.seq_len-1]}\n",
        "        self._active = True\n",
        "\n",
        "        self.client.loop_start()\n",
        "        \n",
        "        # Connect to the Google MQTT bridge.\n",
        "        self.client.connect(self.mqtt_bridge_hostname, self.mqtt_bridge_port)\n",
        "    \n",
        "    def stop(self):\n",
        "        print \"Stop loop\"\n",
        "        self._active = False\n",
        "        self.client.loop_stop()\n",
        "        self.client.disconnect()\n",
        "\n",
        "    @staticmethod\n",
        "    def rotate_message(msg, data, time):\n",
        "        \"\"\"Rotate msg according to the prescribed rule.\n",
        "            Args:\n",
        "             msg: dict,\n",
        "             data: int, active power data.\n",
        "             time: str, the time of measurement\n",
        "            Returns: Payload for use in publishing\n",
        "            \"\"\"\n",
        "        msg['power'].pop(0)\n",
        "        msg['power'].append(data)\n",
        "        msg['timestamp'].pop(0)\n",
        "        msg['timestamp'].append(time)\n",
        "\n",
        "    @staticmethod\n",
        "    def trim_data(data, start_time):\n",
        "        \"\"\"Trims data with start time.\n",
        "            Args:\n",
        "             data: pandas.DataFrame, active power data.\n",
        "             start_time: str, publish data collected from start_time.\n",
        "            Returns: Trimmed data\n",
        "            \"\"\"\n",
        "        if start_time is not None:\n",
        "            print('before data trimming: data.shape={}'.format(data.shape))\n",
        "            data = data[data.index.values >= start_time]\n",
        "            print('after data trimming: data.shape={}'.format(data.shape))\n",
        "        return data\n",
        "\n",
        "    def error_str(self, rc):\n",
        "        \"\"\"Convert a Paho error to a human readable string.\"\"\"\n",
        "        return '{}: {}'.format(rc, mqtt.error_string(rc))\n",
        "\n",
        "    def on_connect(self, unused_client, unused_userdata, unused_flags, rc):\n",
        "        \"\"\"Callback for when a device connects.\"\"\"\n",
        "        print('connected: {}'.format(mqtt.connack_string(rc)))\n",
        "\n",
        "        # After a successful connect, reset backoff time and stop backing off.\n",
        "        Publisher.should_backoff = False\n",
        "        Publisher.minimum_backoff_time = 1\n",
        "\n",
        "        # Trigger initial publish\n",
        "        self.publish()\n",
        "\n",
        "    def on_disconnect(self, unused_client, unused_userdata, rc):\n",
        "        \"\"\"Paho callback for when a device disconnects.\"\"\"\n",
        "        print('disconnected because: {}'.format(error_str(rc)))\n",
        "\n",
        "        # Since a disconnect occurred, the next loop iteration will wait with\n",
        "        # exponential backoff.\n",
        "        Publisher.should_backoff = True\n",
        "\n",
        "    def on_publish(self, unused_client, unused_userdata, unused_mid):\n",
        "        \"\"\"Paho callback when a message is sent to the broker.\"\"\"\n",
        "        time.sleep(self.interval)\n",
        "        self.publish()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "Opz0V0Lrm7D6",
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Load power consumption data\n",
        "\n",
        "power_data = load_data(GOOGLE_CLOUD_PROJECT, POWER_DATA_PATH)\n",
        "print power_data.shape"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "cellView": "form",
        "colab_type": "code",
        "id": "yL1WEVgIm7D-",
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        "# @title Send power data\n",
        "\n",
        "publishers = []\n",
        "for device_id in DEVICE_IDS:\n",
        "    publisher = Publisher(\n",
        "            project_id=GOOGLE_CLOUD_PROJECT,\n",
        "            region=REGION,\n",
        "            registry_id=REGISTRY_ID,\n",
        "            device_id=device_id,\n",
        "            private_key_path=PRIVATE_KEY_PATH,\n",
        "            ca_certs_path=CA_CERTIFICATES_PATH,\n",
        "            jwt_exp_mins=20,\n",
        "            jwt_algorithm='RS256',\n",
        "            mqtt_bridge_hostname='mqtt.googleapis.com',\n",
        "            mqtt_bridge_port=8883,\n",
        "            seq_len=20,\n",
        "            sub_topic='events')\n",
        "\n",
        "    publisher.start(data=power_data, start_time='2013-09-21 20:18:00')\n",
        "    publishers.append(publisher)\n",
        "\n",
        "try: \n",
        "    while True:\n",
        "        time.sleep(1)\n",
        "except KeyboardInterrupt:\n",
        "    for publisher in publishers:\n",
        "        publisher.stop()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "metadata": {
        "colab_type": "code",
        "id": "VtWmsrkUm7ED",
        "colab": {}
      },
      "cell_type": "code",
      "source": [
        ""
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}